package cn.itcast.xc.analysis.subject.salesvolume

import cn.itcast.xc.common.EtlEnvironment
import cn.itcast.xc.entity.{SubjectSaleMonthBiz, SubjectSaleWeekBiz}
import cn.itcast.xc.utils.DateUtils.{getMonthNum, getMonthRelativelyInterval}
import cn.itcast.xc.utils.SubjectUtils.subjectBuyCountSql
import org.apache.spark.sql.SparkSession

/**
 * <p>
 * 学科按月 销售量 统计
 * </p>
 **/
object SubjectSalesVolumeByMonthBiz {
  // 获取spark session
  val spark: SparkSession = EtlEnvironment.getSparkSession(this.getClass.getSimpleName)

  def main(args: Array[String]): Unit = {
    import spark.implicits._
    // 1. 获取当前月的时间区间, 例: 1号 至 18号
    // 2. 获取上月的相同时间区间, 例: 上月1号 至 上月18号
    val date_info = args(0)
//    val date_info = "2019-10-10"
    val months = getMonthRelativelyInterval(date_info).split(",")
    val current = months(0).split(":")
    val last = months(1).split(":")

    // 3. 获取本月的数据
    // 4. 获取上月的数据
    // 5. 计算涨幅
    subjectBuyCountSql(spark, current, last).createOrReplaceTempView("salesvolume_biz_month_tmp")

    // 获取年中的月
    val monthOfYear = getMonthNum(current(0)).split("-")

    // 6. 根据维度id获取相关数据
    val slDS = spark.sql(
      s"""
         |select
         | a.course_category_dim_id,
         | b.name course_category_name,
         | cast(a.sale_num as int ) sale_num,
         | cast(a.sale_percent as float) sale_percent,
         | ${monthOfYear(0)} years,
         | ${monthOfYear(1)} month_of_year
         |from salesvolume_biz_month_tmp a
         |left join data_dimen.course_category_dim b
         |on a.course_category_dim_id = b.course_category_dim_id
         |""".stripMargin).as[SubjectSaleMonthBiz]

    // 7. 保存结果
    // 配置允许动态分区
    spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")
    // 保存到数据仓库
    slDS.repartition(1)
      .write.mode("overwrite")
      .insertInto("data_subject.subject_sale_month_biz")
    // 查看结果
    spark.sql("select * from data_subject.subject_sale_month_biz").show()

    // 关闭资源
    spark.close()
  }

}
